package cn.com.itg.function.process;

import cn.com.itg.pojo.domain.entity.ZTMM;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

public class ZtmmDbConnectCdcFunction extends CoProcessFunction<ZTMM, List<ZTMM>, List<ZTMM>>  implements Serializable {
    // 存储DB过来的数据
    private transient List<ZTMM> ztmms = new ArrayList<>();
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        ztmms = getRuntimeContext()
//                .getListState(new ListStateDescriptor<>("ztmm_db", TypeInformation.of(new TypeHint<ZTMM>() {})))
//        ;
//    }
    @Override
    public void processElement1(ZTMM ztmm, Context context, Collector<List<ZTMM>> collector) throws Exception {
        List<ZTMM> update = ztmms.stream().map(z -> {
                    if (ztmm.getOpType().equals("Update")
                            && Objects.equals(ztmm.getCredId(), z.getCredId())) {
                        return ztmm;
                    }
                    return z;
                }
        ).collect(Collectors.toList());
        if(ztmm.getOpType().equals("Insert")){
            update.add(ztmm);
        }
        ztmms = update;
        collector.collect(update);
    }

    @Override
    public void processElement2(List<ZTMM> ztmmList, Context context, Collector<List<ZTMM>> collector) throws Exception {
        ztmms = ztmmList;
        collector.collect(ztmms);
    }
}
